package com.zytech.zspider.eventbus.rabbitmq;

import com.google.common.collect.ImmutableSet;
import com.rabbitmq.client.*;
import com.zytech.modules.lang.Langs;
import com.zytech.modules.lang.Requires;
import com.zytech.zspider.eventbus.EventEnvelope;
import com.zytech.zspider.eventbus.EventListener;
import com.zytech.zspider.eventbus.support.AutoRetryEventListenerContainer;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * @author siuming
 */
final class RabbitMQEventListenerContainer extends AutoRetryEventListenerContainer {

    private final ConnectionFactory connectionFactory;

    /**
     * @param connectionFactory
     */
    public RabbitMQEventListenerContainer(ConnectionFactory connectionFactory) {
        this(Executors.newScheduledThreadPool(Langs.mycpus()), connectionFactory);
    }

    /**
     * @param scheduledService
     * @param connectionFactory
     */
    public RabbitMQEventListenerContainer(ScheduledExecutorService scheduledService, ConnectionFactory connectionFactory) {
        super(
                scheduledService,
                ImmutableSet.<Class<? extends Exception>>of(
                        IOException.class,
                        ConnectException.class
                )
        );

        Requires.notNull(connectionFactory, "connectionFactory must not be null.");
        this.connectionFactory = connectionFactory;
        this.connectionFactory.setAutomaticRecoveryEnabled(true);
    }

    @Override
    protected void addTopicListenerInternal(String tname, EventListener listener) throws Exception {
        Channel channel = RabbitMQChannels.ofTopicChannel(getConnectionFactory(), tname);
        String queue = RabbitMQChannels.ofTopicChannelQueue(channel, tname);
        channel.basicConsume(queue, new RabbitMQEventListenerAdapter(channel, listener));
    }

    @Override
    protected void addQueueListenerInternal(String qname, EventListener listener) throws Exception {
        Channel channel = RabbitMQChannels.ofQueueChannel(getConnectionFactory(), qname);
        channel.basicConsume(qname, new RabbitMQEventListenerAdapter(channel, listener));
    }

    /**
     * @return
     */
    protected final ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }

    class RabbitMQEventListenerAdapter extends DefaultConsumer {
        final EventListener listener;

        /**
         * @param channel
         * @param listener
         */
        RabbitMQEventListenerAdapter(Channel channel, EventListener listener) {
            super(channel);
            this.listener = listener;
        }

        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            listener.onEvent(ofEventBy(body));
        }

        private EventEnvelope ofEventBy(byte[] body) {
            ByteArrayInputStream in = new ByteArrayInputStream(body);
            return (EventEnvelope) Langs.loads(in);
        }
    }
}
